#include "streammq.h"

namespace Media
{
	

	void StreamMQ::connect(const std::string& connect_string,
		const std::string& exchange,
		const std::string& queue_name,
		Buffer::RingBuffer<std::shared_ptr<StreamPacket> >* stream_buffer)
	{
		if (p_rabbitmq != nullptr)
		{
			delete p_rabbitmq;
			p_rabbitmq = nullptr;
		}

		p_rabbitmq = new Rabbitmq::Rabbitmq(connect_string,
			exchange, queue_name);

		p_stream_buffer = stream_buffer;

		if (!m_running &&  nullptr!= p_stream_buffer)
		{
			this->start();
		}
		else
		{
			Log::printError("StreamMQ启动线程失败，线程正在运行中，或未传入码流buffer");
		}
	}

	void StreamMQ::run()
	{
		while (m_running)
		{
			if (nullptr != p_stream_buffer && nullptr != p_rabbitmq)
			{
				
				auto node = p_stream_buffer->getHeadNode();
				if (!m_running)
				{
					break;
				}
				/*auto buf = node->getData();
				for (int i = 0; i < node->getLen(); i++)
				{
					std::cout << std::hex << (int)buf[i] << " ";
				}*/
				//std::cout << std::endl << std::endl << std::endl << std::endl;
				if (p_rabbitmq->sendData(node->getData(), node->getLen()))
				{
					// 成功后才pop失败了 不pop
					p_stream_buffer->popHeadNode();
				}
			}
			
		}
	}

	bool StreamMQ::stop()
	{
		if (p_thread != nullptr)
		{
			m_running = false;
			// 发一个空包
			std::shared_ptr<StreamPacket> quit_packet(new StreamPacket());
			p_stream_buffer->pushBack(quit_packet);
			p_thread->join();
			delete p_thread;
			p_thread = nullptr;
		}
		return true;
	}
}
